package com.springboot.component;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 消费用户相关的 rocketMq 消息
 * 注意。我这里是springboot项目，实现了CommandLineRunner接口，目的是为了在启动application之后自动运行consumer
 * 用@Order(1)来标记启动consumer的顺序 使用@Component把这个类丢到spring中进行管理
 */
@Component
@Order(1)
public class MessageConsumer implements CommandLineRunner {
    /**
     * NameServer 地址
     */
    @Value("${rocketmq.producerNamesrvAddr}")
    private String nameSrvAddr;
    /**
     * 消费者组名
     */
    @Value("${rocketmq.producerGroupName}")
    private String consumerGroup;
    /**
     * 消费者组名
     */
    @Value("${rocketmq.test.topic}")
    private String topic;
    /**
     * 消费者组名
     */
    @Value("${rocketmq.test.tag}")
    private String tag;

    @Override
    public void run(String... args){
        System.out.println(">>>>>>>>>>>>>>>服务启动执行，执行加载数据等操作<<<<<<<<<<<<<" + nameSrvAddr );
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameSrvAddr);
        consumer.setMaxReconsumeTimes(2);// 设置消费失败最大重试次数 默认16次
        try {
            consumer.subscribe(topic, tag);
//            consumer.subscribe(topic, "*"); // tag:*,可以监控所有该 topic 下的 tag
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    for (MessageExt msg : msgs) {
                        System.out.println("具体内容: " + new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        System.out.printf("Consumer Started.%n");
    }

    public String getNameSrvAddr() {
        return nameSrvAddr;
    }

    public void setNameSrvAddr(String nameSrvAddr) {
        this.nameSrvAddr = nameSrvAddr;
    }

    public String getConsumerGroup() {
        return consumerGroup;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }
}
